/**
 * Date:2022年1月6日 上午10:58:20
 * Copyright (c) 2022, www.bwbroad.com All Rights Reserved
 */

package com.app.common.core;

import com.app.common.utils.BeanUtilsX;
import com.app.common.utils.Utils;
import com.app.common.vo.DebeziumMessage;
import com.app.common.vo.DebeziumMessage.OpType;

/**
 * Description: 消费者基类 <br/>
 * Date:2022年12月6日 15:58:20 <br/>
 * 
 * @author xuejianxin
 */
public abstract class BaseDebeziumConsumer extends AbstractDebeziumConsumer {
	private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(BaseDebeziumConsumer.class);

	@Override
	protected <T, ID> boolean doConsume(DebeziumMessage msg, IMessageHandler<T, ID> h) {

		if (msg.getOp() == OpType.e) {// 删除后的墓碑消息
			log.info("======recive a dumb message,{}", msg.getKey());
			return true;
		}

		// 构建主键
		var dataKey = h.buildId(msg);// getID(msg);

		// 需要考虑在分区表中的联合主键
		// jpa 中的联合主键 https://www.jianshu.com/p/6ba194fe3674
		var optEn = h.queryByBizID(dataKey);
		var en = optEn.orElse(h.getEmptyEntity());
		// 前置处理
		h.preProcess(msg, en);
		try {

			switch (msg.getOp()) {
			case r:
			case c:
				// 把map中的值合并到en
				BeanUtilsX.mapToBean(en, Utils.underlineToCamel(msg.getAfter()));
				// 有则更新无则插入,为什么都是更新 需要_new 来标识数据是新增
				h.save(en);
				break;
			case u:
				if (optEn.isPresent()) {// 更新时如何实现部分更新?先删后插？部分更新？
					// 把map中的值合并到en
					BeanUtilsX.mapToBean(en, Utils.underlineToCamel(msg.getAfter()));
					// 有则更新无则插入
					h.save(en);
				} else {// 数据不存在，用 after 覆盖before,然后插入 before
					msg.getBefore().putAll(msg.getAfter());
					BeanUtilsX.mapToBean(en, Utils.underlineToCamel(msg.getBefore()));
					h.save(en);
				}
				break;
			case d:// 多数据库复制时是否要删除?
				if (optEn.isPresent()) {
					h.deleteByBizID(dataKey);
				}
				break;

			default:
				break;
			}
			// 后置处理
			h.postProcess(en);
			return true;
		} catch (Exception e) {
			log.error("", e);
			return false;
		}
	}

}
